/*-------------------------------------------------------------------------
 *
 * nbtxlog.c
 *	  btree 的 WAL 重放逻辑。
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/access/nbtree/nbtxlog.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/bufmask.h"
#include "access/nbtree.h"
#include "access/nbtxlog.h"
#include "access/transam.h"
#include "access/xlog.h"
#include "access/xlogutils.h"
#include "miscadmin.h"
#include "storage/procarray.h"
#include "utils/memutils.h"

static MemoryContext opCtx;		/* 工作内存用于操作 */

/*
 * _bt_restore_page -- 重新进入页面上的所有索引元组
 *
 * 该页面是新初始化的，*from（长度len）是其上部分（pd_upper到pd_special）的一个副本。
 * 我们假设元组是按项目编号顺序添加到页面上的，因此具有最高项目编号的元组首先出现（在页面的最低处）。
 */
static void fc__bt_restore_page(Page fc_page, char *fc_from, int fc_len)
{
	IndexTupleData fc_itupdata;
	Size		fc_itemsz;
	char	   *fc_end = fc_from + fc_len;
	Item		fc_items[MaxIndexTuplesPerPage];
	uint16		fc_itemsizes[MaxIndexTuplesPerPage];
	int			fc_i;
	int			fc_nitems;

	/*
	 * 为了以原始顺序恢复这些项目，我们以相反的顺序将它们添加到页面上。
	 * 为了确定一个元组结束和另一个元组开始的位置，我们必须首先按前向顺序扫描它们。
	 */
	fc_i = 0;
	while (fc_from < fc_end)
	{
		/*
		 * 当我们逐步处理这些项目时，'from' 并不总是正确对齐，因此我们需要使用 memcpy()。
		 * 此外，我们在此处使用 Item（它只是一个 char*）作为我们的项目数组，原因是相同的；
		 * 不希望编译器或其他人认为一个项目是对齐的，而实际上并不是。
		 */
		memcpy(&fc_itupdata, fc_from, sizeof(IndexTupleData));
		fc_itemsz = IndexTupleSize(&fc_itupdata);
		fc_itemsz = MAXALIGN(fc_itemsz);

		fc_items[fc_i] = (Item) fc_from;
		fc_itemsizes[fc_i] = fc_itemsz;
		fc_i++;

		fc_from += fc_itemsz;
	}
	fc_nitems = fc_i;

	for (fc_i = fc_nitems - 1; fc_i >= 0; fc_i--)
	{
		if (PageAddItem(fc_page, fc_items[fc_i], fc_itemsizes[fc_i], fc_nitems - fc_i,
						false, false) == InvalidOffsetNumber)
			elog(PANIC, "_bt_restore_page: cannot add item to page");
	}
}

static void fc__bt_restore_meta(XLogReaderState *fc_record, uint8 fc_block_id)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	Buffer		fc_metabuf;
	Page		fc_metapg;
	BTMetaPageData *fc_md;
	BTPageOpaque fc_pageop;
	xl_btree_metadata *fc_xlrec;
	char	   *fc_ptr;
	Size		fc_len;

	fc_metabuf = XLogInitBufferForRedo(fc_record, fc_block_id);
	fc_ptr = XLogRecGetBlockData(fc_record, fc_block_id, &fc_len);

	Assert(fc_len == sizeof(xl_btree_metadata));
	Assert(BufferGetBlockNumber(fc_metabuf) == BTREE_METAPAGE);
	fc_xlrec = (xl_btree_metadata *) fc_ptr;
	fc_metapg = BufferGetPage(fc_metabuf);

	_bt_pageinit(fc_metapg, BufferGetPageSize(fc_metabuf));

	fc_md = BTPageGetMeta(fc_metapg);
	fc_md->btm_magic = BTREE_MAGIC;
	fc_md->btm_version = fc_xlrec->version;
	fc_md->btm_root = fc_xlrec->root;
	fc_md->btm_level = fc_xlrec->level;
	fc_md->btm_fastroot = fc_xlrec->fastroot;
	fc_md->btm_fastlevel = fc_xlrec->fastlevel;
	/* 无法在升级前记录 BTREE_MIN_VERSION 索引元数据页 */
	Assert(fc_md->btm_version >= BTREE_NOVAC_VERSION);
	fc_md->btm_last_cleanup_num_delpages = fc_xlrec->last_cleanup_num_delpages;
	fc_md->btm_last_cleanup_num_heap_tuples = -1.0;
	fc_md->btm_allequalimage = fc_xlrec->allequalimage;

	fc_pageop = BTPageGetOpaque(fc_metapg);
	fc_pageop->btpo_flags = BTP_META;

	/*
	 * 将 pd_lower 设置在元数据结束的后面。这是至关重要的，因为不这样做，如果 xlog.c 压缩页面，元数据将丢失。
	 */
	((PageHeader) fc_metapg)->pd_lower =
		((char *) fc_md + sizeof(BTMetaPageData)) - (char *) fc_metapg;

	PageSetLSN(fc_metapg, fc_lsn);
	MarkBufferDirty(fc_metabuf);
	UnlockReleaseBuffer(fc_metabuf);
}

/*
 * _bt_clear_incomplete_split -- 清除页面上的 INCOMPLETE_SPLIT 标志
 *
 * 这是所有能够插入下行链接的 WAL 记录类型的重做函数的一个常见子例程：插入、拆分和新根。
 */
static void fc__bt_clear_incomplete_split(XLogReaderState *fc_record, uint8 fc_block_id)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	Buffer		fc_buf;

	if (XLogReadBufferForRedo(fc_record, fc_block_id, &fc_buf) == BLK_NEEDS_REDO)
	{
		Page		fc_page = (Page) BufferGetPage(fc_buf);
		BTPageOpaque fc_pageop = BTPageGetOpaque(fc_page);

		Assert(P_INCOMPLETE_SPLIT(fc_pageop));
		fc_pageop->btpo_flags &= ~BTP_INCOMPLETE_SPLIT;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buf);
	}
	if (BufferIsValid(fc_buf))
		UnlockReleaseBuffer(fc_buf);
}

static void fc_btree_xlog_insert(bool fc_isleaf, bool fc_ismeta, bool fc_posting,
				  XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_insert *fc_xlrec = (xl_btree_insert *) XLogRecGetData(fc_record);
	Buffer		fc_buffer;
	Page		fc_page;

	/*
	 * 对内部页面的插入在子级完成不完整的拆分。
	 * 清除子级中的不完整拆分标志。注意：在正常操作期间，子级和父级页面是同时锁定的
	 * （锁是耦合的），因此清除标志和插入下行链接在其他后端看来是原子的。
	 * 在重播期间我们不管这个，因为读取者不关心不完整拆分标志，且不能有更新发生。
	 */
	if (!fc_isleaf)
		fc__bt_clear_incomplete_split(fc_record, 1);
	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		Size		fc_datalen;
		char	   *fc_datapos = XLogRecGetBlockData(fc_record, 0, &fc_datalen);

		fc_page = BufferGetPage(fc_buffer);

		if (!fc_posting)
		{
			/* 简单的零售插入 */
			if (PageAddItem(fc_page, (Item) fc_datapos, fc_datalen, fc_xlrec->offnum,
							false, false) == InvalidOffsetNumber)
				elog(PANIC, "failed to add new item");
		}
		else
		{
			ItemId		fc_itemid;
			IndexTuple	fc_oposting,
						fc_newitem,
						fc_nposting;
			uint16		fc_postingoff;

			/*
			 * 在叶页面插入期间发生了一个发布列表拆分。WAL
			 * 记录数据将以一个偏移量编号开始，表示发生拆分的现有发布列表中的点。
			 *
			 * 使用 _bt_swap_posting() 从主记录重复发布列表拆分步骤。
			 * 注意，来自 WAL 记录的新项目是 'orignewitem'，
			 * 不是实际插入页面的最终版本的新项目。
			 */
			fc_postingoff = *((uint16 *) fc_datapos);
			fc_datapos += sizeof(uint16);
			fc_datalen -= sizeof(uint16);

			fc_itemid = PageGetItemId(fc_page, OffsetNumberPrev(fc_xlrec->offnum));
			fc_oposting = (IndexTuple) PageGetItem(fc_page, fc_itemid);

			/* 在 _bt_swap_posting() 中使用可变的、对齐的新项目副本 */
			Assert(fc_isleaf && fc_postingoff > 0);
			fc_newitem = CopyIndexTuple((IndexTuple) fc_datapos);
			fc_nposting = _bt_swap_posting(fc_newitem, fc_oposting, fc_postingoff);

			/* 用拆分后的版本替换现有发布列表 */
			memcpy(fc_oposting, fc_nposting, MAXALIGN(IndexTupleSize(fc_nposting)));

			/* 插入“最终”的新项目（不是来自 WAL 流的 orignewitem） */
			Assert(IndexTupleSize(fc_newitem) == fc_datalen);
			if (PageAddItem(fc_page, (Item) fc_newitem, fc_datalen, fc_xlrec->offnum,
							false, false) == InvalidOffsetNumber)
				elog(PANIC, "failed to add posting split new item");
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/*
	 * 注意：在正常操作中，我们将在仍然保持对插入页面的锁的同时更新元数据页。
	 * 但在重播期间不需要保持该锁，因为没有其他索引更新可能同时发生，
	 * 而读取者会很好地处理从元数据页跟随一个过时的链接。
	 */
	if (fc_ismeta)
		fc__bt_restore_meta(fc_record, 2);
}

static void fc_btree_xlog_split(bool fc_newitemonleft, XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_split *fc_xlrec = (xl_btree_split *) XLogRecGetData(fc_record);
	bool		fc_isleaf = (fc_xlrec->level == 0);
	Buffer		fc_buf;
	Buffer		fc_rbuf;
	Page		fc_rpage;
	BTPageOpaque fc_ropaque;
	char	   *fc_datapos;
	Size		fc_datalen;
	BlockNumber fc_origpagenumber;
	BlockNumber fc_rightpagenumber;
	BlockNumber fc_spagenumber;

	XLogRecGetBlockTag(fc_record, 0, NULL, NULL, &fc_origpagenumber);
	XLogRecGetBlockTag(fc_record, 1, NULL, NULL, &fc_rightpagenumber);
	if (!XLogRecGetBlockTagExtended(fc_record, 2, NULL, NULL, &fc_spagenumber, NULL))
		fc_spagenumber = P_NONE;

	/*
	 * 清除适当子页面一层下的不完整拆分标志，当origpage/buf是内部页面时（在内部页面拆分的情况下，必须有级联页面拆分发生）。这像是对应于内部页面的btree_xlog_insert调用。我们此处并没有清除当前页面拆分的不完整拆分标志（你可以将其视为页面拆分操作在传递中新项插入的一部分）。
	 *
	 * 像在btree_xlog_insert中一样，这可以在锁定其他页面之前完成。我们在REDO例程中不需要耦合跨级别锁定。
	 */
	if (!fc_isleaf)
		fc__bt_clear_incomplete_split(fc_record, 3);

	/* 从头开始重建右侧（新）兄弟页面 */
	fc_rbuf = XLogInitBufferForRedo(fc_record, 1);
	fc_datapos = XLogRecGetBlockData(fc_record, 1, &fc_datalen);
	fc_rpage = (Page) BufferGetPage(fc_rbuf);

	_bt_pageinit(fc_rpage, BufferGetPageSize(fc_rbuf));
	fc_ropaque = BTPageGetOpaque(fc_rpage);

	fc_ropaque->btpo_prev = fc_origpagenumber;
	fc_ropaque->btpo_next = fc_spagenumber;
	fc_ropaque->btpo_level = fc_xlrec->level;
	fc_ropaque->btpo_flags = fc_isleaf ? BTP_LEAF : 0;
	fc_ropaque->btpo_cycleid = 0;

	fc__bt_restore_page(fc_rpage, fc_datapos, fc_datalen);

	PageSetLSN(fc_rpage, fc_lsn);
	MarkBufferDirty(fc_rbuf);

	/* 现在重建原始页面（拆分的左半部分） */
	if (XLogReadBufferForRedo(fc_record, 0, &fc_buf) == BLK_NEEDS_REDO)
	{
		/*
		 * 为了保留元组的相同物理顺序，我们为左侧页面初始化一个临时空页面，并按项目编号顺序将所有项添加到其中。这与_bt_split()的工作方式相符。保留相同的物理顺序使WAL一致性检查成为可能。另请参见_bt_restore_page()，它对右侧页面执行相同操作。
		 */
		Page		fc_origpage = (Page) BufferGetPage(fc_buf);
		BTPageOpaque fc_oopaque = BTPageGetOpaque(fc_origpage);
		OffsetNumber fc_off;
		IndexTuple	fc_newitem = NULL,
					fc_left_hikey = NULL,
					fc_nposting = NULL;
		Size		fc_newitemsz = 0,
					fc_left_hikeysz = 0;
		Page		fc_leftpage;
		OffsetNumber fc_leftoff,
					fc_replacepostingoff = InvalidOffsetNumber;

		fc_datapos = XLogRecGetBlockData(fc_record, 0, &fc_datalen);

		if (fc_newitemonleft || fc_xlrec->postingoff != 0)
		{
			fc_newitem = (IndexTuple) fc_datapos;
			fc_newitemsz = MAXALIGN(IndexTupleSize(fc_newitem));
			fc_datapos += fc_newitemsz;
			fc_datalen -= fc_newitemsz;

			if (fc_xlrec->postingoff != 0)
			{
				ItemId		fc_itemid;
				IndexTuple	fc_oposting;

				/* 发布列表必须位于新项之前的偏移量号 */
				fc_replacepostingoff = OffsetNumberPrev(fc_xlrec->newitemoff);

				/* 在 _bt_swap_posting() 中使用可变的、对齐的新项目副本 */
				fc_newitem = CopyIndexTuple(fc_newitem);
				fc_itemid = PageGetItemId(fc_origpage, fc_replacepostingoff);
				fc_oposting = (IndexTuple) PageGetItem(fc_origpage, fc_itemid);
				fc_nposting = _bt_swap_posting(fc_newitem, fc_oposting,
											fc_xlrec->postingoff);
			}
		}

		/*
		 * 提取左侧高键及其大小。我们假设16位对齐足以应用IndexTupleSize（因为它是从uint16字段中获取的）。
		 */
		fc_left_hikey = (IndexTuple) fc_datapos;
		fc_left_hikeysz = MAXALIGN(IndexTupleSize(fc_left_hikey));
		fc_datapos += fc_left_hikeysz;
		fc_datalen -= fc_left_hikeysz;

		Assert(fc_datalen == 0);

		fc_leftpage = PageGetTempPageCopySpecial(fc_origpage);

		/* 将高键元组从WAL记录添加到临时页面 */
		fc_leftoff = P_HIKEY;
		if (PageAddItem(fc_leftpage, (Item) fc_left_hikey, fc_left_hikeysz, P_HIKEY,
						false, false) == InvalidOffsetNumber)
			elog(ERROR, "failed to add high key to left page after split");
		fc_leftoff = OffsetNumberNext(fc_leftoff);

		for (fc_off = P_FIRSTDATAKEY(fc_oopaque); fc_off < fc_xlrec->firstrightoff; fc_off++)
		{
			ItemId		fc_itemid;
			Size		fc_itemsz;
			IndexTuple	fc_item;

			/* 在需要时添加替换的发布列表 */
			if (fc_off == fc_replacepostingoff)
			{
				Assert(fc_newitemonleft ||
					   fc_xlrec->firstrightoff == fc_xlrec->newitemoff);
				if (PageAddItem(fc_leftpage, (Item) fc_nposting,
								MAXALIGN(IndexTupleSize(fc_nposting)), fc_leftoff,
								false, false) == InvalidOffsetNumber)
					elog(ERROR, "failed to add new posting list item to left page after split");
				fc_leftoff = OffsetNumberNext(fc_leftoff);
				continue;		/* 不插入oposting */
			}

			/* 如果新项已插入左侧页面，则添加新项 */
			else if (fc_newitemonleft && fc_off == fc_xlrec->newitemoff)
			{
				if (PageAddItem(fc_leftpage, (Item) fc_newitem, fc_newitemsz, fc_leftoff,
								false, false) == InvalidOffsetNumber)
					elog(ERROR, "failed to add new item to left page after split");
				fc_leftoff = OffsetNumberNext(fc_leftoff);
			}

			fc_itemid = PageGetItemId(fc_origpage, fc_off);
			fc_itemsz = ItemIdGetLength(fc_itemid);
			fc_item = (IndexTuple) PageGetItem(fc_origpage, fc_itemid);
			if (PageAddItem(fc_leftpage, (Item) fc_item, fc_itemsz, fc_leftoff,
							false, false) == InvalidOffsetNumber)
				elog(ERROR, "failed to add old item to left page after split");
			fc_leftoff = OffsetNumberNext(fc_leftoff);
		}

		/* 处理新项可能位于末尾的情况 */
		if (fc_newitemonleft && fc_off == fc_xlrec->newitemoff)
		{
			if (PageAddItem(fc_leftpage, (Item) fc_newitem, fc_newitemsz, fc_leftoff,
							false, false) == InvalidOffsetNumber)
				elog(ERROR, "failed to add new item to left page after split");
			fc_leftoff = OffsetNumberNext(fc_leftoff);
		}

		PageRestoreTempPage(fc_leftpage, fc_origpage);

		/* 修复不透明字段 */
		fc_oopaque->btpo_flags = BTP_INCOMPLETE_SPLIT;
		if (fc_isleaf)
			fc_oopaque->btpo_flags |= BTP_LEAF;
		fc_oopaque->btpo_next = fc_rightpagenumber;
		fc_oopaque->btpo_cycleid = 0;

		PageSetLSN(fc_origpage, fc_lsn);
		MarkBufferDirty(fc_buf);
	}

	/* 修复新右侧兄弟页面右侧页面的左链接 */
	if (fc_spagenumber != P_NONE)
	{
		Buffer		fc_sbuf;

		if (XLogReadBufferForRedo(fc_record, 2, &fc_sbuf) == BLK_NEEDS_REDO)
		{
			Page		fc_spage = (Page) BufferGetPage(fc_sbuf);
			BTPageOpaque fc_spageop = BTPageGetOpaque(fc_spage);

			fc_spageop->btpo_prev = fc_rightpagenumber;

			PageSetLSN(fc_spage, fc_lsn);
			MarkBufferDirty(fc_sbuf);
		}
		if (BufferIsValid(fc_sbuf))
			UnlockReleaseBuffer(fc_sbuf);
	}

	/*
	 * 最后，释放剩余的缓存区。sbuf、rbuf和buf必须一起释放，以便读者无法观察到不一致性。
	 */
	UnlockReleaseBuffer(fc_rbuf);
	if (BufferIsValid(fc_buf))
		UnlockReleaseBuffer(fc_buf);
}

static void fc_btree_xlog_dedup(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_dedup *fc_xlrec = (xl_btree_dedup *) XLogRecGetData(fc_record);
	Buffer		fc_buf;

	if (XLogReadBufferForRedo(fc_record, 0, &fc_buf) == BLK_NEEDS_REDO)
	{
		char	   *fc_ptr = XLogRecGetBlockData(fc_record, 0, NULL);
		Page		fc_page = (Page) BufferGetPage(fc_buf);
		BTPageOpaque fc_opaque = BTPageGetOpaque(fc_page);
		OffsetNumber fc_offnum,
					fc_minoff,
					fc_maxoff;
		BTDedupState fc_state;
		BTDedupInterval *fc_intervals;
		Page		fc_newpage;

		fc_state = (BTDedupState) palloc(sizeof(BTDedupStateData));
		fc_state->deduplicate = true;	/* 未使用 */
		fc_state->nmaxitems = 0;	/* 未使用 */
		/* 保守地使用比主节点更大的maxpostingsize */
		fc_state->maxpostingsize = BTMaxItemSize(fc_page);
		fc_state->base = NULL;
		fc_state->baseoff = InvalidOffsetNumber;
		fc_state->basetupsize = 0;
		fc_state->htids = palloc(fc_state->maxpostingsize);
		fc_state->nhtids = 0;
		fc_state->nitems = 0;
		fc_state->phystupsize = 0;
		fc_state->nintervals = 0;

		fc_minoff = P_FIRSTDATAKEY(fc_opaque);
		fc_maxoff = PageGetMaxOffsetNumber(fc_page);
		fc_newpage = PageGetTempPageCopySpecial(fc_page);

		if (!P_RIGHTMOST(fc_opaque))
		{
			ItemId		fc_itemid = PageGetItemId(fc_page, P_HIKEY);
			Size		fc_itemsz = ItemIdGetLength(fc_itemid);
			IndexTuple	fc_item = (IndexTuple) PageGetItem(fc_page, fc_itemid);

			if (PageAddItem(fc_newpage, (Item) fc_item, fc_itemsz, P_HIKEY,
							false, false) == InvalidOffsetNumber)
				elog(ERROR, "deduplication failed to add highkey");
		}

		fc_intervals = (BTDedupInterval *) fc_ptr;
		for (fc_offnum = fc_minoff;
			 fc_offnum <= fc_maxoff;
			 fc_offnum = OffsetNumberNext(fc_offnum))
		{
			ItemId		fc_itemid = PageGetItemId(fc_page, fc_offnum);
			IndexTuple	fc_itup = (IndexTuple) PageGetItem(fc_page, fc_itemid);

			if (fc_offnum == fc_minoff)
				_bt_dedup_start_pending(fc_state, fc_itup, fc_offnum);
			else if (fc_state->nintervals < fc_xlrec->nintervals &&
					 fc_state->baseoff == fc_intervals[fc_state->nintervals].baseoff &&
					 fc_state->nitems < fc_intervals[fc_state->nintervals].nitems)
			{
				if (!_bt_dedup_save_htid(fc_state, fc_itup))
					elog(ERROR, "deduplication failed to add heap tid to pending posting list");
			}
			else
			{
				_bt_dedup_finish_pending(fc_newpage, fc_state);
				_bt_dedup_start_pending(fc_state, fc_itup, fc_offnum);
			}
		}

		_bt_dedup_finish_pending(fc_newpage, fc_state);
		Assert(fc_state->nintervals == fc_xlrec->nintervals);
		Assert(memcmp(fc_state->intervals, fc_intervals,
					  fc_state->nintervals * sizeof(BTDedupInterval)) == 0);

		if (P_HAS_GARBAGE(fc_opaque))
		{
			BTPageOpaque fc_nopaque = BTPageGetOpaque(fc_newpage);

			fc_nopaque->btpo_flags &= ~BTP_HAS_GARBAGE;
		}

		PageRestoreTempPage(fc_newpage, fc_page);
		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buf);
	}

	if (BufferIsValid(fc_buf))
		UnlockReleaseBuffer(fc_buf);
}

static void fc_btree_xlog_updates(Page fc_page, OffsetNumber *fc_updatedoffsets,
				   xl_btree_update *fc_updates, int fc_nupdated)
{
	BTVacuumPosting fc_vacposting;
	IndexTuple	fc_origtuple;
	ItemId		fc_itemid;
	Size		fc_itemsz;

	for (int fc_i = 0; fc_i < fc_nupdated; fc_i++)
	{
		fc_itemid = PageGetItemId(fc_page, fc_updatedoffsets[fc_i]);
		fc_origtuple = (IndexTuple) PageGetItem(fc_page, fc_itemid);

		fc_vacposting = palloc(offsetof(BTVacuumPostingData, deletetids) +
							fc_updates->ndeletedtids * sizeof(uint16));
		fc_vacposting->updatedoffset = fc_updatedoffsets[fc_i];
		fc_vacposting->itup = fc_origtuple;
		fc_vacposting->ndeletedtids = fc_updates->ndeletedtids;
		memcpy(fc_vacposting->deletetids,
			   (char *) fc_updates + SizeOfBtreeUpdate,
			   fc_updates->ndeletedtids * sizeof(uint16));

		_bt_update_posting(fc_vacposting);

		/* 重写元组的更新版本 */
		fc_itemsz = MAXALIGN(IndexTupleSize(fc_vacposting->itup));
		if (!PageIndexTupleOverwrite(fc_page, fc_updatedoffsets[fc_i],
									 (Item) fc_vacposting->itup, fc_itemsz))
			elog(PANIC, "failed to update partially dead item");

		pfree(fc_vacposting->itup);
		pfree(fc_vacposting);

		/* 移动到数组中的下一个xl_btree_update */
		fc_updates = (xl_btree_update *)
			((char *) fc_updates + SizeOfBtreeUpdate +
			 fc_updates->ndeletedtids * sizeof(uint16));
	}
}

static void fc_btree_xlog_vacuum(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_vacuum *fc_xlrec = (xl_btree_vacuum *) XLogRecGetData(fc_record);
	Buffer		fc_buffer;
	Page		fc_page;
	BTPageOpaque fc_opaque;

	/*
	 * 我们在这里需要获取一个清理锁，就像btvacuum页面()那样。然而，在恢复期间，没有必要在索引中的每个块上彻底获取清理锁（只需在有待删除项的页面上获取清理锁即可）。有关详细信息，请参见nbtree/README。
	 */
	if (XLogReadBufferForRedoExtended(fc_record, 0, RBM_NORMAL, true, &fc_buffer)
		== BLK_NEEDS_REDO)
	{
		char	   *fc_ptr = XLogRecGetBlockData(fc_record, 0, NULL);

		fc_page = (Page) BufferGetPage(fc_buffer);

		if (fc_xlrec->nupdated > 0)
		{
			OffsetNumber *fc_updatedoffsets;
			xl_btree_update *fc_updates;

			fc_updatedoffsets = (OffsetNumber *)
				(fc_ptr + fc_xlrec->ndeleted * sizeof(OffsetNumber));
			fc_updates = (xl_btree_update *) ((char *) fc_updatedoffsets +
										   fc_xlrec->nupdated *
										   sizeof(OffsetNumber));

			fc_btree_xlog_updates(fc_page, fc_updatedoffsets, fc_updates, fc_xlrec->nupdated);
		}

		if (fc_xlrec->ndeleted > 0)
			PageIndexMultiDelete(fc_page, (OffsetNumber *) fc_ptr, fc_xlrec->ndeleted);

		/*
		 * 标记页面为不包含任何LP_DEAD项 --- 参见_bt_delitems_vacuum()中的评论。
		 */
		fc_opaque = BTPageGetOpaque(fc_page);
		fc_opaque->btpo_flags &= ~BTP_HAS_GARBAGE;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

static void fc_btree_xlog_delete(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_delete *fc_xlrec = (xl_btree_delete *) XLogRecGetData(fc_record);
	Buffer		fc_buffer;
	Page		fc_page;
	BTPageOpaque fc_opaque;

	/*
	 * 如果我们有任何冲突处理需要执行，它必须在我们更新页面之前发生。
	 */
	if (InHotStandby)
	{
		RelFileNode fc_rnode;

		XLogRecGetBlockTag(fc_record, 0, &fc_rnode, NULL, NULL);

		ResolveRecoveryConflictWithSnapshot(fc_xlrec->latestRemovedXid, fc_rnode);
	}

	/*
	 * 我们不需要获取清理锁来应用这些更改。有关详细信息，请参见nbtree/README。
	 */
	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		char	   *fc_ptr = XLogRecGetBlockData(fc_record, 0, NULL);

		fc_page = (Page) BufferGetPage(fc_buffer);

		if (fc_xlrec->nupdated > 0)
		{
			OffsetNumber *fc_updatedoffsets;
			xl_btree_update *fc_updates;

			fc_updatedoffsets = (OffsetNumber *)
				(fc_ptr + fc_xlrec->ndeleted * sizeof(OffsetNumber));
			fc_updates = (xl_btree_update *) ((char *) fc_updatedoffsets +
										   fc_xlrec->nupdated *
										   sizeof(OffsetNumber));

			fc_btree_xlog_updates(fc_page, fc_updatedoffsets, fc_updates, fc_xlrec->nupdated);
		}

		if (fc_xlrec->ndeleted > 0)
			PageIndexMultiDelete(fc_page, (OffsetNumber *) fc_ptr, fc_xlrec->ndeleted);

		/* 标记页面为不包含任何LP_DEAD项 */
		fc_opaque = BTPageGetOpaque(fc_page);
		fc_opaque->btpo_flags &= ~BTP_HAS_GARBAGE;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

static void fc_btree_xlog_mark_page_halfdead(uint8 fc_info, XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_mark_page_halfdead *fc_xlrec = (xl_btree_mark_page_halfdead *) XLogRecGetData(fc_record);
	Buffer		fc_buffer;
	Page		fc_page;
	BTPageOpaque fc_pageop;
	IndexTupleData fc_trunctuple;

	/*
	 * 在正常操作中，我们会在更改任何与此WAL记录相关的页面之前锁定所有页面。 
	 * 在WAL重放中，逐次锁定一页是可以的，因为不会有并发的索引更新发生， 
	 * 读者不应该关心他们是否到达目标页面（因为它肯定是空的）。
	 */

	/* 待删除子树的父页面 */
	if (XLogReadBufferForRedo(fc_record, 1, &fc_buffer) == BLK_NEEDS_REDO)
	{
		OffsetNumber fc_poffset;
		ItemId		fc_itemid;
		IndexTuple	fc_itup;
		OffsetNumber fc_nextoffset;
		BlockNumber fc_rightsib;

		fc_page = (Page) BufferGetPage(fc_buffer);
		fc_pageop = BTPageGetOpaque(fc_page);

		fc_poffset = fc_xlrec->poffset;

		fc_nextoffset = OffsetNumberNext(fc_poffset);
		fc_itemid = PageGetItemId(fc_page, fc_nextoffset);
		fc_itup = (IndexTuple) PageGetItem(fc_page, fc_itemid);
		fc_rightsib = BTreeTupleGetDownLink(fc_itup);

		fc_itemid = PageGetItemId(fc_page, fc_poffset);
		fc_itup = (IndexTuple) PageGetItem(fc_page, fc_itemid);
		BTreeTupleSetDownLink(fc_itup, fc_rightsib);
		fc_nextoffset = OffsetNumberNext(fc_poffset);
		PageIndexTupleDelete(fc_page, fc_nextoffset);

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}

	/*
	 * 在REDO例程中不需要耦合跨层级锁，因此立即释放
	 * 内部页面上的锁
	 */
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* 将叶子页面重写为半死页面 */
	fc_buffer = XLogInitBufferForRedo(fc_record, 0);
	fc_page = (Page) BufferGetPage(fc_buffer);

	_bt_pageinit(fc_page, BufferGetPageSize(fc_buffer));
	fc_pageop = BTPageGetOpaque(fc_page);

	fc_pageop->btpo_prev = fc_xlrec->leftblk;
	fc_pageop->btpo_next = fc_xlrec->rightblk;
	fc_pageop->btpo_level = 0;
	fc_pageop->btpo_flags = BTP_HALF_DEAD | BTP_LEAF;
	fc_pageop->btpo_cycleid = 0;

	/*
	 * 构造一个指向顶部父页面的虚拟高键项（当顶级父页面本身是叶子页面时，值为InvalidBlockNumber）
	 */
	MemSet(&fc_trunctuple, 0, sizeof(IndexTupleData));
	fc_trunctuple.t_info = sizeof(IndexTupleData);
	BTreeTupleSetTopParent(&fc_trunctuple, fc_xlrec->topparent);

	if (PageAddItem(fc_page, (Item) &fc_trunctuple, sizeof(IndexTupleData), P_HIKEY,
					false, false) == InvalidOffsetNumber)
		elog(ERROR, "could not add dummy high key to half-dead page");

	PageSetLSN(fc_page, fc_lsn);
	MarkBufferDirty(fc_buffer);
	UnlockReleaseBuffer(fc_buffer);
}


static void fc_btree_xlog_unlink_page(uint8 fc_info, XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_unlink_page *fc_xlrec = (xl_btree_unlink_page *) XLogRecGetData(fc_record);
	BlockNumber fc_leftsib;
	BlockNumber fc_rightsib;
	uint32		fc_level;
	bool		fc_isleaf;
	FullTransactionId fc_safexid;
	Buffer		fc_leftbuf;
	Buffer		fc_target;
	Buffer		fc_rightbuf;
	Page		fc_page;
	BTPageOpaque fc_pageop;

	fc_leftsib = fc_xlrec->leftsib;
	fc_rightsib = fc_xlrec->rightsib;
	fc_level = fc_xlrec->level;
	fc_isleaf = (fc_level == 0);
	fc_safexid = fc_xlrec->safexid;

	/* 对于0级（叶子页面）或1级目标，没有父叶子页面 */
	Assert(!BlockNumberIsValid(fc_xlrec->leaftopparent) || fc_level > 1);

	/*
	 * 在正常操作中，我们会在更改任何与此WAL记录相关的页面之前锁定所有页面。 
	 * 在WAL重放中，我们至少按照标准的从左到右顺序锁定页面（左兄弟、目标、右兄弟）， 
	 * 并且不会在目标标记为删除之前释放兄弟锁。
	 */

	/* 修复左兄弟的右链接，如果有的话 */
	if (fc_leftsib != P_NONE)
	{
		if (XLogReadBufferForRedo(fc_record, 1, &fc_leftbuf) == BLK_NEEDS_REDO)
		{
			fc_page = (Page) BufferGetPage(fc_leftbuf);
			fc_pageop = BTPageGetOpaque(fc_page);
			fc_pageop->btpo_next = fc_rightsib;

			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_leftbuf);
		}
	}
	else
		fc_leftbuf = InvalidBuffer;

	/* 将目标页面重写为空的删除页面 */
	fc_target = XLogInitBufferForRedo(fc_record, 0);
	fc_page = (Page) BufferGetPage(fc_target);

	_bt_pageinit(fc_page, BufferGetPageSize(fc_target));
	fc_pageop = BTPageGetOpaque(fc_page);

	fc_pageop->btpo_prev = fc_leftsib;
	fc_pageop->btpo_next = fc_rightsib;
	fc_pageop->btpo_level = fc_level;
	BTPageSetDeleted(fc_page, fc_safexid);
	if (fc_isleaf)
		fc_pageop->btpo_flags |= BTP_LEAF;
	fc_pageop->btpo_cycleid = 0;

	PageSetLSN(fc_page, fc_lsn);
	MarkBufferDirty(fc_target);

	/* 修复右兄弟的左链接 */
	if (XLogReadBufferForRedo(fc_record, 2, &fc_rightbuf) == BLK_NEEDS_REDO)
	{
		fc_page = (Page) BufferGetPage(fc_rightbuf);
		fc_pageop = BTPageGetOpaque(fc_page);
		fc_pageop->btpo_prev = fc_leftsib;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_rightbuf);
	}

	/* 释放兄弟 */
	if (BufferIsValid(fc_leftbuf))
		UnlockReleaseBuffer(fc_leftbuf);
	if (BufferIsValid(fc_rightbuf))
		UnlockReleaseBuffer(fc_rightbuf);

	/* 释放目标 */
	UnlockReleaseBuffer(fc_target);

	/*
	 * 如果我们删除了目标叶子页面的父页面，而不是叶子本身，
	 * 更新叶子以指向待删除子树中下一个剩余的子节点
	 */
	if (XLogRecHasBlockRef(fc_record, 3))
	{
		/*
		 * 页面上没有真实数据，因此我们只需根据
		 * WAL记录中的信息从头开始重新创建它。
		 *
		 * 请注意，当目标页面也是叶子缓冲区页面时，我们不会到达这里。
		 * 删除叶子缓冲区时无需添加带有顶级父链接的虚拟高键项，
		 * 因为它是我们将删除的子树中最后一个页面。
		 */
		Buffer		fc_leafbuf;
		IndexTupleData fc_trunctuple;

		Assert(!fc_isleaf);

		fc_leafbuf = XLogInitBufferForRedo(fc_record, 3);
		fc_page = (Page) BufferGetPage(fc_leafbuf);

		_bt_pageinit(fc_page, BufferGetPageSize(fc_leafbuf));
		fc_pageop = BTPageGetOpaque(fc_page);

		fc_pageop->btpo_flags = BTP_HALF_DEAD | BTP_LEAF;
		fc_pageop->btpo_prev = fc_xlrec->leafleftsib;
		fc_pageop->btpo_next = fc_xlrec->leafrightsib;
		fc_pageop->btpo_level = 0;
		fc_pageop->btpo_cycleid = 0;

		/* 添加一个虚拟高键项 */
		MemSet(&fc_trunctuple, 0, sizeof(IndexTupleData));
		fc_trunctuple.t_info = sizeof(IndexTupleData);
		BTreeTupleSetTopParent(&fc_trunctuple, fc_xlrec->leaftopparent);

		if (PageAddItem(fc_page, (Item) &fc_trunctuple, sizeof(IndexTupleData), P_HIKEY,
						false, false) == InvalidOffsetNumber)
			elog(ERROR, "could not add dummy high key to half-dead page");

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_leafbuf);
		UnlockReleaseBuffer(fc_leafbuf);
	}

	/* 如有需要，更新元页面 */
	if (fc_info == XLOG_BTREE_UNLINK_PAGE_META)
		fc__bt_restore_meta(fc_record, 4);
}

static void fc_btree_xlog_newroot(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_btree_newroot *fc_xlrec = (xl_btree_newroot *) XLogRecGetData(fc_record);
	Buffer		fc_buffer;
	Page		fc_page;
	BTPageOpaque fc_pageop;
	char	   *fc_ptr;
	Size		fc_len;

	fc_buffer = XLogInitBufferForRedo(fc_record, 0);
	fc_page = (Page) BufferGetPage(fc_buffer);

	_bt_pageinit(fc_page, BufferGetPageSize(fc_buffer));
	fc_pageop = BTPageGetOpaque(fc_page);

	fc_pageop->btpo_flags = BTP_ROOT;
	fc_pageop->btpo_prev = fc_pageop->btpo_next = P_NONE;
	fc_pageop->btpo_level = fc_xlrec->level;
	if (fc_xlrec->level == 0)
		fc_pageop->btpo_flags |= BTP_LEAF;
	fc_pageop->btpo_cycleid = 0;

	if (fc_xlrec->level > 0)
	{
		fc_ptr = XLogRecGetBlockData(fc_record, 0, &fc_len);
		fc__bt_restore_page(fc_page, fc_ptr, fc_len);

		/* 清除左子节点中的不完整分裂标志 */
		fc__bt_clear_incomplete_split(fc_record, 1);
	}

	PageSetLSN(fc_page, fc_lsn);
	MarkBufferDirty(fc_buffer);
	UnlockReleaseBuffer(fc_buffer);

	fc__bt_restore_meta(fc_record, 2);
}

/*
 * 一般来说，VACUUM必须推迟回收，以避免某些竞争条件。已删除页包含一个safexid值，VACUUM使用它来确定现在将已删除的页面安全地放入FSM是否安全。请参见nbtree/README。
 *
 * 就任何在原始执行期间运行的后端而言，FSM是一个可回收安全页面的缓存；页面在FSM中的存在仅表示该页面必须已经安全可回收（实际上，_bt_getbuf()通过BTPageIsRecyclable()验证其安全，但这仅仅是因为考虑到当前的限制，完全信任FSM是不明智的）。
 *
 * 然而，这不足以防止在热备份期间发生类似的并发回收竞争条件。为此，我们需要在页面实际被回收并用于完全不相关的页面的过程中，在_bt_split()中记录一个xl_btree_reuse_page记录。这些记录包含来自原始已删除页面的相同safexid值，存储在记录的latestRemovedFullXid字段中。
 *
 * BTPageIsRecyclable()中的GlobalVisCheckRemovableFullXid()测试用于确定是否安全地回收一个页面。这与我们自己的测试相对应：PGPROC->xmin > limitXmin测试在GetConflictingVirtualXIDs()中。因此，一个XID值在主库和备库上达成相同的排除效果。
 */
static void fc_btree_xlog_reuse_page(XLogReaderState *fc_record)
{
	xl_btree_reuse_page *fc_xlrec = (xl_btree_reuse_page *) XLogRecGetData(fc_record);

	if (InHotStandby)
		ResolveRecoveryConflictWithSnapshotFullXid(fc_xlrec->latestRemovedFullXid,
												   fc_xlrec->node);
}

void btree_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;
	MemoryContext fc_oldCtx;

	fc_oldCtx = MemoryContextSwitchTo(opCtx);
	switch (fc_info)
	{
		case XLOG_BTREE_INSERT_LEAF:
			fc_btree_xlog_insert(true, false, false, fc_record);
			break;
		case XLOG_BTREE_INSERT_UPPER:
			fc_btree_xlog_insert(false, false, false, fc_record);
			break;
		case XLOG_BTREE_INSERT_META:
			fc_btree_xlog_insert(false, true, false, fc_record);
			break;
		case XLOG_BTREE_SPLIT_L:
			fc_btree_xlog_split(true, fc_record);
			break;
		case XLOG_BTREE_SPLIT_R:
			fc_btree_xlog_split(false, fc_record);
			break;
		case XLOG_BTREE_INSERT_POST:
			fc_btree_xlog_insert(true, false, true, fc_record);
			break;
		case XLOG_BTREE_DEDUP:
			fc_btree_xlog_dedup(fc_record);
			break;
		case XLOG_BTREE_VACUUM:
			fc_btree_xlog_vacuum(fc_record);
			break;
		case XLOG_BTREE_DELETE:
			fc_btree_xlog_delete(fc_record);
			break;
		case XLOG_BTREE_MARK_PAGE_HALFDEAD:
			fc_btree_xlog_mark_page_halfdead(fc_info, fc_record);
			break;
		case XLOG_BTREE_UNLINK_PAGE:
		case XLOG_BTREE_UNLINK_PAGE_META:
			fc_btree_xlog_unlink_page(fc_info, fc_record);
			break;
		case XLOG_BTREE_NEWROOT:
			fc_btree_xlog_newroot(fc_record);
			break;
		case XLOG_BTREE_REUSE_PAGE:
			fc_btree_xlog_reuse_page(fc_record);
			break;
		case XLOG_BTREE_META_CLEANUP:
			fc__bt_restore_meta(fc_record, 0);
			break;
		default:
			elog(PANIC, "btree_redo: unknown op code %u", fc_info);
	}
	MemoryContextSwitchTo(fc_oldCtx);
	MemoryContextReset(opCtx);
}

void btree_xlog_startup(void)
{
	opCtx = AllocSetContextCreate(CurrentMemoryContext,
								  "Btree recovery temporary context",
								  ALLOCSET_DEFAULT_SIZES);
}

void btree_xlog_cleanup(void)
{
	MemoryContextDelete(opCtx);
	opCtx = NULL;
}

/*
 * 在对btree页面执行一致性检查之前，对其进行掩码处理。
 */
void btree_mask(char *fc_pagedata, BlockNumber fc_blkno)
{
	Page		fc_page = (Page) fc_pagedata;
	BTPageOpaque fc_maskopaq;

	mask_page_lsn_and_checksum(fc_page);

	mask_page_hint_bits(fc_page);
	mask_unused_space(fc_page);

	fc_maskopaq = BTPageGetOpaque(fc_page);

	if (P_ISLEAF(fc_maskopaq))
	{
		/*
		 * 在btree叶子页面中，可以在不发出任何WAL记录的情况下修改LP_FLAGS。因此，掩码行指针标志。有关详细信息，请参见_bt_killitems()、_bt_check_unique()。
		 */
		mask_lp_flags(fc_page);
	}

	/*
	 * BTP_HAS_GARBAGE只是一个未记录的提示位。因此，掩码它。有关详细信息，请参见_bt_delete_or_dedup_one_page()、_bt_killitems()和_bt_check_unique()。
	 */
	fc_maskopaq->btpo_flags &= ~BTP_HAS_GARBAGE;

	/*
	 * 在btree页面分裂的重播期间，我们不为右兄弟设置BTP_SPLIT_END标志，并将相同页面的cycle_id初始化为0。有关详细信息，请参见btree_xlog_split()。
	 */
	fc_maskopaq->btpo_flags &= ~BTP_SPLIT_END;
	fc_maskopaq->btpo_cycleid = 0;
}
